package com.atguigu.sparkcore.day02.kv

import org.apache.spark.{SparkConf, SparkContext}

/**
 * Author atguigu
 * Date 2020/10/28 14:05
 */
object AggregateByKeyDemo {
    def main(args: Array[String]): Unit = {
        val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")
        
        //2.创建SparkContext，该对象是提交Spark App的入口
        val sc: SparkContext = new SparkContext(conf)
        
        //3具体业务逻辑
        //3.1 创建第一个RDD
        val rdd1 = sc.makeRDD(List(("a", 1), ("b", 5), ("a", 2), ("a", 7), ("a", 5), ("b", 2)), 2)
        //        val rdd2 = rdd1.aggregateByKey(Int.MinValue)((x, y) => x.max(y), (x, y) => x + y)
        //        val rdd2 = rdd1.aggregateByKey(Int.MaxValue)((x, y) => x.min(y), (x, y) => x + y)
        // 一次性计算出来每个分区内最大和最小的和
        val rdd2 = rdd1.aggregateByKey((Int.MinValue, Int.MaxValue))(
            {
                case ((max, min), v) =>
                    (max.max(v), min.min(v))
            },
            {
                case ((max1, min1), (max2, min2)) =>
                    (max1+max2, min1 + min2)
            }
        )
        
        rdd2.collect.foreach(println)
        
        sc.stop()
    }
}
/*
aggregateByKey
    有零值
    分区内和分区间的聚合逻辑不一样
foldByKey
    有零值
    分区内和分区间的聚合逻辑一样
reduceByKey
    没有零值
    分区内和分区间的聚合逻辑一样


 */
